package com.pandy.disruptor.senior;

import com.lmax.disruptor.BusySpinWaitStrategy;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @description: 高级特性：并行计算
 *
 * 单消费者：底层使用的是 BatchEventProcessor
 *
 * @author: panhong
 * @date: 2025/4/10
 */
public class Main {

    public static void main(String[] args) throws Exception {

        ExecutorService executorService = Executors.newFixedThreadPool(5);
        ExecutorService es = Executors.newFixedThreadPool(4);

        // 创建 disruptor
        Disruptor<Goods> disruptor = new Disruptor<>(
                new EventFactory<Goods>() {
                    @Override
                    public Goods newInstance() {
                        return new Goods();
                    }
                },
                1024 * 1024,
                executorService,
                ProducerType.SINGLE,
                new BusySpinWaitStrategy());

        // 2把消费者添加到disruptor中 handleEventsWith

        // 2.1 串行操作 使用链式调用
//        disruptor.handleEventsWith(new Handler1()).handleEventsWith(new Handler2()).handleEventsWith(new Handler3());

        // 2.2 并行操作 使用单独调用
//        disruptor.handleEventsWith(new Handler1(),new Handler2(),new Handler3());
//        disruptor.handleEventsWith(new Handler1());
//        disruptor.handleEventsWith(new Handler2());
//        disruptor.handleEventsWith(new Handler3());

        // 2.3 多边形操作
        // 2.3.1 菱形操作
        /**
         * 1 ---- 2 ---- 4
         *   -----3
         */
        // 1 2,3 4 串行操作，2 3 并行操作
//       disruptor.handleEventsWith(new Handler1()).handleEventsWith(new Handler2(),new Handler3())
//               .handleEventsWith(new Handler4());

        // 2.3.2 六边形操作
        // 这里要注意 和disruptor中的线程池个数设置一致 (原因后面源码理解) -- 有几个Handler就要几个线程
        // ExecutorService executorService = Executors.newFixedThreadPool(5);
        Handler1 h1 = new Handler1();
        Handler2 h2 = new Handler2();
        Handler3 h3 = new Handler3();
        Handler4 h4 = new Handler4();
        Handler5 h5 = new Handler5();
        disruptor.handleEventsWith(h1, h4);
        disruptor.after(h1).handleEventsWith(h2);
        disruptor.after(h4).handleEventsWith(h5);
        disruptor.after(h2, h5).handleEventsWith(h3);

        // 启动disruptor
        RingBuffer<Goods> ringBuffer = disruptor.start();

        long begin = System.currentTimeMillis();

        CountDownLatch latch = new CountDownLatch(1);
        es.submit(new GoodsPublisher(disruptor, latch));

        latch.await();

        // 关闭
        disruptor.shutdown();
        executorService.shutdown();
        es.shutdown();

        System.out.println("总共用时：" + (System.currentTimeMillis() - begin));
    }
}
